package com.zhangdi.mserver.goes.core;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.zhangdi.mserver.goes.config.MonitorOption;

import java.net.InetSocketAddress;
import java.net.SocketAddress;

/**
 * Created by zhangdi on 2021/01/29 14:30:31.
 */
public class CanalMonitorFactory {
    private final int BATCH_SIZE = 5 * 1024;
    private final String FILTER = "*.\\..*";
    private final String CANAL_USERNAME = "";
    private final String CANAL_PASSWORD = "";
    private final int EMPTY_DATA_IDLE = 10*1000;

    public Monitor builder(String hostname, int port, String destination) {
        MonitorOption option = new MonitorOption();
        option.setHostname(hostname);
        option.setPort(port);
        option.setDestination(destination);
        option.setUsername(CANAL_USERNAME);
        option.setPassword(CANAL_PASSWORD);
        option.setSubscribe(FILTER);
        option.setBatchSize(BATCH_SIZE);
        option.setEmptyIdle(EMPTY_DATA_IDLE);

        return builder(option);
    }

    public Monitor builder(String hostname, int port, String destination, String filter) {
        MonitorOption option = new MonitorOption();
        option.setHostname(hostname);
        option.setPort(port);
        option.setDestination(destination);
        option.setUsername(CANAL_USERNAME);
        option.setPassword(CANAL_PASSWORD);
        option.setBatchSize(BATCH_SIZE);
        option.setEmptyIdle(EMPTY_DATA_IDLE);
        option.setSubscribe(filter);

        return builder(option);
    }

    public Monitor builder(MonitorOption option) {
        return new CanalMonitor(option, canalConnector(option));
    }

    private CanalConnector canalConnector(MonitorOption option) {
        SocketAddress addressInfo = new InetSocketAddress(option.getHostname(), option.getPort());
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(addressInfo, option.getDestination(),
                option.getUsername(), option.getPassword());
        return canalConnector;
    }
}
